package eventBusV61

import (
	"errors"
	"io"
	"log"
	"strings"
	"sync"
	"time"
)

const (
	pubTimeout = time.Millisecond * 10
)

var (
	ErrPubsubTimeout = errors.New("failed to send message to topic because of timeout")
)

type Pubsub interface {
	Publish(topic string, msg interface{}) error
	Subscribe(topic string) (<-chan interface{}, error)
	Unsubscribe(topic string, ch <-chan interface{}) error
	io.Closer
}

type pubsub struct {
	size     int
	channels sync.Map // 使用 sync.Map 已经是线程安全的，不需要额外的锁。
	log      *log.Logger
}

func NewPubsub(size int) (Pubsub, error) {
	return &pubsub{
		size: size,
		log:  log.New(io.Discard, "pubsub: ", log.LstdFlags),
	}, nil
}

func (m *pubsub) Publish(topic string, msg interface{}) error {
	var errs []string
	if chs, ok := m.getChannel(topic); ok {
		for _, ch := range chs {
			err := m.publish(ch, msg)
			if err != nil {
				errs = append(errs, err.Error())
			}
		}
	}
	if len(errs) > 0 {
		return errors.New(strings.Join(errs, "\n"))
	}
	return nil
}

func (m *pubsub) Subscribe(topic string) (<-chan interface{}, error) {
	ch := make(chan interface{}, m.size) // 使用带缓冲的通道

	chs, _ := m.channels.LoadOrStore(topic, map[<-chan interface{}]chan interface{}{})
	chs.(map[<-chan interface{}]chan interface{})[ch] = ch
	return ch, nil
}

func (m *pubsub) Unsubscribe(topic string, ch <-chan interface{}) error {
	if chs, ok := m.channels.Load(topic); ok {
		delete(chs.(map[<-chan interface{}]chan interface{}), ch)
	}
	return nil
}

func (m *pubsub) Close() error {
	m.channels.Range(func(key, value interface{}) bool {
		for k := range value.(map[<-chan interface{}]chan interface{}) {
			delete(value.(map[<-chan interface{}]chan interface{}), k)
		}
		m.channels.Delete(key)
		return true
	})
	return nil
}

func (m *pubsub) publish(ch chan interface{}, msg interface{}) error {
	timer := time.NewTimer(pubTimeout)
	defer timer.Stop()

	select {
	case ch <- msg:
	case <-timer.C:
		m.log.Println("publish message timeout")
		return ErrPubsubTimeout
	}
	return nil
}

func (m *pubsub) getChannel(topic string) (map[<-chan interface{}]chan interface{}, bool) {
	if chs, ok := m.channels.Load(topic); ok {
		return chs.(map[<-chan interface{}]chan interface{}), true
	}
	return nil, false
}
